package com.test.rxjava;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import lombok.extern.slf4j.Slf4j;


@Slf4j
public class RxjavaDemo {

    public static void main(String[] args) {
        // 创建被观察者
        Observable observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                log.info("=========================currentThread name: " + Thread.currentThread().getName());
//                e.onNext(1);
//                e.onNext(2);
//                e.onNext(3);
                e.onComplete();
            }
        });
        // 创建观察者
        Observer observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                log.info("======================onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                log.info("======================onNext " + integer);
            }

            @Override
            public void onError(Throwable e) {
                log.info("======================onError");
            }

            @Override
            public void onComplete() {
                log.info("======================onComplete");
            }
        };
        // 订阅
        observable
                .just(1, 2, 3)
                .subscribe(observer);
    }
}
